package com.yoyo.schedule.zookeeper;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.yoyo.schedule.Constants;
import com.yoyo.schedule.Schedules;
import com.yoyo.schedule.TaskItemAndParameter;
import com.yoyo.schedule.Uuids;
import com.yoyo.schedule.task_manager.*;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.sql.Timestamp;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Slf4j
public class DataManager implements IDataManager {
    @Setter
    @Getter
    private Gson gson;
    @Setter
    @Getter
    private ZooKeeperManager zooKeeperManager;
    @Setter
    @Getter
    private String baseTaskTypePath;
    @Setter
    @Getter
    private String taskItems = "taskItems";
    @Setter
    @Getter
    private String servers = "servers";
    @Setter
    @Getter
    private long zooKeeperBaseDateTimeLong = 0;
    @Setter
    @Getter
    private long localBaseDateTimeLong = 0;

    public DataManager(ZooKeeperManager zooKeeperManager) throws Exception {
        this.zooKeeperManager = zooKeeperManager;
        this.localBaseDateTimeLong = System.currentTimeMillis();
        this.gson = new GsonBuilder().registerTypeAdapter(Timestamp.class, new TimestampTypeAdapter()).setDateFormat("yyyy-MM-dd HH:mm:ss").create();

        this.baseTaskTypePath = this.zooKeeperManager.getRootPath() + Constants.$base_task_type;
        if (this.getZooKeeper().exists(this.baseTaskTypePath, false) == null) {
            ZooKeepers.createPath(getZooKeeper(), this.baseTaskTypePath, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }

        //
        String systemTimePath = this.zooKeeperManager.getZooKeeper().create(this.zooKeeperManager.getRootPath() + Constants.$system_time, null, this.zooKeeperManager.getAclList(), CreateMode.EPHEMERAL_SEQUENTIAL);
        Stat stat = this.zooKeeperManager.getZooKeeper().exists(systemTimePath, false);
        this.zooKeeperBaseDateTimeLong = stat.getCtime();
        //
        ZooKeepers.directoryStructureDelete(getZooKeeper(), systemTimePath);
        if (Math.abs(this.zooKeeperBaseDateTimeLong - this.localBaseDateTimeLong) > 5000) {
            log.error("Zookeeper " + Math.abs(this.zooKeeperBaseDateTimeLong - this.localBaseDateTimeLong) + " ms");
        }
    }

    public ZooKeeper getZooKeeper() throws Exception {
        return this.zooKeeperManager.getZooKeeper();
    }

    //
    public void createBaseTaskType(TaskType baseTaskType) throws Exception {
        if (baseTaskType.getBaseTaskType().indexOf(Constants.$) > 0)//baseTaskType不能包含$
        {
            throw new IllegalArgumentException("baseTaskType不能包含$");
        }
        String baseTaskTypePath = this.baseTaskTypePath + Constants.slash + baseTaskType.getBaseTaskType();
        String value = this.gson.toJson(baseTaskType);
        if (this.getZooKeeper().exists(baseTaskTypePath, false) == null) {
            this.getZooKeeper().create(baseTaskTypePath, value.getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        } else {
            throw new IllegalArgumentException("baseTaskType已经存在");
        }
    }

    public void updateBaseTaskType(TaskType taskType)
            throws Exception {
        if (taskType.getBaseTaskType().indexOf(Constants.$) > 0)//baseTaskType不能包含$
        {
            throw new IllegalArgumentException("baseTaskType不能包含$");
        }
        String baseTaskTypePath = this.baseTaskTypePath + Constants.slash + taskType.getBaseTaskType();
        String value = this.gson.toJson(taskType);
        if (this.getZooKeeper().exists(baseTaskTypePath, false) == null) {
            this.getZooKeeper().create(baseTaskTypePath, value.getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        } else {
            this.getZooKeeper().setData(baseTaskTypePath, value.getBytes(), -1);
        }

    }

    //动态
    public void createTaskType(String baseTaskType, String ownSign) throws Exception {
        String taskType = Schedules.getTaskType(baseTaskType, ownSign);
        String taskTypePath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType;
        if (this.getZooKeeper().exists(taskTypePath, false) == null) {
            this.getZooKeeper().create(taskTypePath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
    }

    //静态
    public void createTaskItems(String baseTaskType, String ownSign, String serverId)
            throws Exception {

        String taskType = Schedules.getTaskType(baseTaskType, ownSign);
        String taskItemsPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems;
        try {
            ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskItemsPath);
        } catch (Exception e) {
            if (e instanceof KeeperException
                    && ((KeeperException) e).code().intValue() == KeeperException.Code.SESSIONEXPIRED.intValue()) {
                log.warn("delete : zookeeper sessionzookeeper");
                zooKeeperManager.reConnect();
                ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskItemsPath);
            }
        }
        this.getZooKeeper().create(taskItemsPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        this.createTaskItems(baseTaskType, ownSign, this.loadBaseTaskType(baseTaskType).getTaskItems());
        setTaskItemsWithServerId(baseTaskType, taskType, serverId);//核心关键点
    }

    public void setTaskItemsWithServerId(String baseTaskType, String taskType, String serverId) throws Exception {
        String taskItemsPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems;
        this.getZooKeeper().setData(taskItemsPath, serverId.getBytes(), -1);
    }

    public boolean isLeaderServer(String baseTaskType, String ownSign) throws Exception {
        String taskType = Schedules.getTaskType(baseTaskType, ownSign);
        String leaderServerId = this.getLeaderServerId(this.loadServerIdList(taskType));
        String taskItemsPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems;
        if (this.getZooKeeper().exists(taskItemsPath, false) != null) {
            byte[] bytes = this.getZooKeeper().getData(taskItemsPath, false, null);
            return bytes != null && new String(bytes).equals(leaderServerId);
        }
        return false;
    }

    //
    public long updateTaskItemFlag(String taskType) throws Exception {
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String serversPath = this.baseTaskTypePath + Constants.slash + baseTaskType
                + Constants.slash + taskType + Constants.slash + this.servers;
        Stat stat = this.getZooKeeper().setData(serversPath, "reload=true".getBytes(), -1);
        return stat.getVersion();

    }

    public Map<String, Stat> getServerIdToStatMap(String taskType) throws Exception {
        Map<String, Stat> serverIdToStatMap = new HashMap<String, Stat>();
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String serversPath = this.baseTaskTypePath + Constants.slash + baseTaskType
                + Constants.slash + taskType + Constants.slash + this.servers;
        List<String> serverIdList = this.getZooKeeper().getChildren(serversPath, false);
        for (String serverId : serverIdList) {
            String serverIdPath = serversPath + Constants.slash + serverId;
            Stat stat = this.getZooKeeper().exists(serverIdPath, false);
            serverIdToStatMap.put(serverId, stat);
        }
        return serverIdToStatMap;
    }

    public long getTaskItemStatVersion(String taskType) throws Exception {
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String serversPath = this.baseTaskTypePath + Constants.slash + baseTaskType
                + Constants.slash + taskType + Constants.slash + this.servers;
        Stat stat = new Stat();
        this.getZooKeeper().getData(serversPath, false, stat);
        return stat.getVersion();
    }

    public void createTaskItems(String baseTaskType, String ownSign, String[] baseTaskItems) throws Exception {
        TaskItem[] taskItems = new TaskItem[baseTaskItems.length];
        Pattern pattern = Pattern.compile("\\s*:\\s*\\{");

        for (int i = 0; i < baseTaskItems.length; i++) {
            taskItems[i] = new TaskItem();
            taskItems[i].setBaseTaskType(baseTaskType);
            taskItems[i].setOwnSign(ownSign);
            taskItems[i].setTaskType(Schedules.getTaskType(baseTaskType, ownSign));
            //
            Matcher matcher = pattern.matcher(baseTaskItems[i]);
            if (matcher.find()) {
                taskItems[i].setTaskItemString(baseTaskItems[i].substring(0, matcher.start()).trim());
                taskItems[i].setDealParameter(baseTaskItems[i].substring(matcher.end(), baseTaskItems[i].length() - 1).trim());
            } else {
                taskItems[i].setTaskItemString(baseTaskItems[i]);
            }
            taskItems[i].setStatus(Status.ACTIVTE);
        }
        createTaskItemIfNotFind(taskItems);
    }

    public void createTaskItemIfNotFind(TaskItem[] taskItems) throws Exception {
        for (TaskItem taskItem : taskItems) {
            String taskItemsPath = this.baseTaskTypePath + Constants.slash + taskItem.getBaseTaskType() + Constants.slash + taskItem.getTaskType() + Constants.slash + this.taskItems;
            if (this.getZooKeeper().exists(taskItemsPath, false) == null) {
                ZooKeepers.createPath(this.getZooKeeper(), taskItemsPath, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            }
            String taskItemPath = taskItemsPath + Constants.slash + taskItem.getTaskItemString();
            this.getZooKeeper().create(taskItemPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            this.getZooKeeper().create(taskItemPath + Constants.$current_server_id, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            this.getZooKeeper().create(taskItemPath + Constants.$request_server_id, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            this.getZooKeeper().create(taskItemPath + Constants.$status, taskItem.getStatus().toString().getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            this.getZooKeeper().create(taskItemPath + Constants.$parameter, taskItem.getDealParameter().getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            this.getZooKeeper().create(taskItemPath + Constants.$deal_info, taskItem.getDealInfo().getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
    }

    public void updateTaskItemStatus(String taskType, String taskItem, Status status, String message) throws Exception {
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String taskItemPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems + Constants.slash + taskItem;
        if (this.getZooKeeper().exists(taskItemPath + Constants.$sts, false) == null) {
            this.getZooKeeper().setData(taskItemPath + Constants.$sts, status.toString().getBytes(), -1);
        }
        if (this.getZooKeeper().exists(taskItemPath + Constants.$deal_info, false) == null) {
            if (message == null) {
                message = "";
            }
            this.getZooKeeper().setData(taskItemPath + Constants.$deal_info, message.getBytes(), -1);
        }
    }

    public void deleteTaskItem(String taskType, String taskItem) throws Exception {
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String taskItemPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems + Constants.slash + taskItem;
        ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskItemPath);
    }

    public List<TaskItem> loadTaskItemList(String taskType) throws Exception {
        List<TaskItem> taskItemList = new ArrayList<TaskItem>();
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String taskItemsPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems;
        if (this.getZooKeeper().exists(taskItemsPath, false) == null) {
            return taskItemList;
        }
        List<String> taskItemStringList = this.getZooKeeper().getChildren(taskItemsPath, false);
        Collections.sort(taskItemStringList, new Comparator<String>() {
            public int compare(String leftString, String rightString) {
                if (StringUtils.isNumeric(leftString) && StringUtils.isNumeric(rightString)) {
                    int leftIntString = Integer.parseInt(leftString);
                    int rightIntString = Integer.parseInt(rightString);
                    if (leftIntString == rightIntString) {
                        return 0;
                    } else if (leftIntString > rightIntString) {
                        return 1;
                    } else {
                        return -1;
                    }
                } else {
                    return leftString.compareTo(rightString);
                }
            }
        });// end
        for (String taskItemString : taskItemStringList) {
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskType(taskType);
            taskItem.setTaskItemString(taskItemString);
            String taskItemPath = taskItemsPath + Constants.slash + taskItemString;
            byte[] currentServerIdBytes = this.getZooKeeper().getData(taskItemPath + Constants.$current_server_id, false, null);
            if (currentServerIdBytes != null) {
                taskItem.setCurrentServerId(new String(currentServerIdBytes));
            }
            byte[] requestServerIdBytes = this.getZooKeeper().getData(taskItemPath + Constants.$request_server_id, false, null);
            if (requestServerIdBytes != null) {
                taskItem.setRequestServerId(new String(requestServerIdBytes));
            }
            byte[] statusBytes = this.getZooKeeper().getData(taskItemPath + Constants.$status, false, null);
            if (statusBytes != null) {
                taskItem.setStatus(Status.valueOf(new String(statusBytes)));
            }
            byte[] parameterBytes = this.getZooKeeper().getData(taskItemPath + Constants.$parameter, false, null);
            if (parameterBytes != null) {
                taskItem.setDealParameter(new String(parameterBytes));
            }
            byte[] dealInfoBytes = this.getZooKeeper().getData(taskItemPath + Constants.$deal_info, false, null);
            if (dealInfoBytes != null) {
                taskItem.setDealInfo(new String(dealInfoBytes));
            }
            taskItemList.add(taskItem);
        }
        return taskItemList;

    }

    //
    public TaskType loadBaseTaskType(String baseTaskType)
            throws Exception {
        String baseTaskTypePath = this.baseTaskTypePath + Constants.slash + baseTaskType;
        if (this.getZooKeeper().exists(baseTaskTypePath, false) == null) {
            return null;
        }
        String value = new String(this.getZooKeeper().getData(baseTaskTypePath, false, null));
        TaskType taskType = this.gson.fromJson(value, TaskType.class);
        return taskType;
    }


    public List<TaskType> getBaseTaskTypeList() throws Exception {
        String baseTaskTypePath = this.baseTaskTypePath;
        List<TaskType> taskTypeList = new ArrayList<TaskType>();
        List<String> baseTaskTypeList = this.getZooKeeper().getChildren(baseTaskTypePath, false);
        Collections.sort(baseTaskTypeList);
        for (String baseTaskType : baseTaskTypeList) {
            taskTypeList.add(this.loadBaseTaskType(baseTaskType));
        }
        return taskTypeList;
    }


    public void clearTaskType(String baseTaskType) throws Exception {
        String baseTaskTypePath = this.baseTaskTypePath + Constants.slash + baseTaskType;
        List<String> taskTypeList = this.getZooKeeper().getChildren(baseTaskTypePath, false);
        for (String taskType : taskTypeList) {
            ZooKeepers.directoryStructureDelete(this.getZooKeeper(), baseTaskTypePath + Constants.slash + taskType);
        }
    }


    public List<TaskTypeRuntimeInfo> getTaskTypeRunInfoList(
            String baseTaskType) throws Exception {
        List<TaskTypeRuntimeInfo> taskTypeRuntimeInfoList = new ArrayList<TaskTypeRuntimeInfo>();
        String baseTaskTypePath = this.baseTaskTypePath + Constants.slash + baseTaskType;
        if (this.getZooKeeper().exists(baseTaskTypePath, false) == null) {
            return taskTypeRuntimeInfoList;
        }
        List<String> taskTypeList = this.getZooKeeper().getChildren(baseTaskTypePath, false);
        Collections.sort(taskTypeList);

        for (String taskType : taskTypeList) {
            TaskTypeRuntimeInfo taskTypeRuntimeInfo = new TaskTypeRuntimeInfo();
            taskTypeRuntimeInfo.setBaseTaskType(baseTaskType);
            taskTypeRuntimeInfo.setTaskType(taskType);
            taskTypeRuntimeInfo.setOwnSign(Schedules.getOwnSign(taskType));
            taskTypeRuntimeInfoList.add(taskTypeRuntimeInfo);
        }
        return taskTypeRuntimeInfoList;
    }


    public void deleteTaskType(String baseTaskType) throws Exception {
        ZooKeepers.directoryStructureDelete(this.getZooKeeper(), this.baseTaskTypePath + Constants.slash + baseTaskType);
    }


    public List<TaskHandler> getTaskHandlerList(
            String baseTaskType,//
            String ownSign,//
            String ip, //
            String orderString) throws Exception {//
        List<String> taskTypeList = new ArrayList<String>();
        if (baseTaskType != null && ownSign != null)//两个都不为null
        {
            taskTypeList.add(baseTaskType + "$" + ownSign);
        } else if (baseTaskType != null && ownSign == null)//baseTaskType != null && ownSign == null
        {
            if (this.getZooKeeper().exists(this.baseTaskTypePath + Constants.slash + baseTaskType, false) != null) {
                for (String taskType : this.getZooKeeper().getChildren(this.baseTaskTypePath + Constants.slash + baseTaskType, false)) {
                    taskTypeList.add(taskType);
                }
            }
        } else if (baseTaskType == null)//baseTaskType == null
        {
            for (String taskType : this.getZooKeeper().getChildren(this.baseTaskTypePath, false)) {
                if (ownSign != null)//ownSign != null
                {
                    taskTypeList.add(taskType + "$" + ownSign);
                } else//ownSign == null
                {
                    for (String taskTypeString : this.getZooKeeper().getChildren(this.baseTaskTypePath + Constants.slash + taskType, false)) {
                        taskTypeList.add(taskTypeString);
                    }
                }
            }
        }
        //
        List<TaskHandler> taskHandlerList = new ArrayList<TaskHandler>();
        for (String taskType : taskTypeList) {
            List<TaskHandler> temporaryTaskHandlerList = this.getTaskHandlerList(taskType);
            if (ip == null) {
                taskHandlerList.addAll(temporaryTaskHandlerList);
            } else {
                for (TaskHandler taskHandler : temporaryTaskHandlerList) {
                    if (ip.equals(taskHandler.getIp()))//ip限制
                    {
                        taskHandlerList.add(taskHandler);
                    }
                }
            }
        }
        Collections.sort(taskHandlerList, new TaskHandlerComparator(orderString));

        return taskHandlerList;
    }



    public List<TaskItemAndParameter> getTaskItemAndParameterList(String taskType, String serverId)
            throws Exception {
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String taskItemsPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems;

        List<String> taskItemList = this.getZooKeeper().getChildren(taskItemsPath, false);
        Collections.sort(taskItemList, new Comparator<String>() {
            public int compare(String leftString, String rightString) {
                if (StringUtils.isNumeric(leftString) && StringUtils.isNumeric(rightString)) {
                    int leftIntValue = Integer.parseInt(leftString);
                    int rightIntValue = Integer.parseInt(rightString);
                    if (leftIntValue == rightIntValue) {
                        return 0;
                    } else if (leftIntValue > rightIntValue) {
                        return 1;
                    } else {
                        return -1;
                    }
                } else {
                    return leftString.compareTo(rightString);
                }
            }
        });//end

        log.debug(taskType + " current uid=" + serverId + " , zookeeper  getTaskItemAndParameterList");

        List<TaskItemAndParameter> taskItemAndParameterList = new ArrayList<TaskItemAndParameter>();
        for (String taskItem : taskItemList) {
            byte[] currentServerIdBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItem + Constants.$current_server_id, false, null);

            if (currentServerIdBytes != null && serverId.equals(new String(currentServerIdBytes))) {
                TaskItemAndParameter taskItemAndParameter = new TaskItemAndParameter();
                taskItemAndParameter.setTaskItem(taskItem);//taskItem/parameter
                byte[] parameterBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItem + Constants.$parameter, false, null);
                if (parameterBytes != null) {
                    taskItemAndParameter.setParameter(new String(parameterBytes));
                }
                taskItemAndParameterList.add(taskItemAndParameter);


            } else if (currentServerIdBytes != null && serverId.equals(new String(currentServerIdBytes)) == false) {
                log.trace(" current uid=" + serverId + " , zookeeper cur_server uid=" + new String(currentServerIdBytes));
            } else {
                log.trace(" current uid=" + serverId);
            }
        }
        return taskItemAndParameterList;
    }


    public void releaseDealTaskItem(String taskType, String serverId) throws Exception {
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String taskItemsPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems;
        boolean isModify = false;
        for (String taskItem : this.getZooKeeper().getChildren(taskItemsPath, false)) {
            byte[] currentServerIdBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItem + Constants.$current_server_id, false, null);
            byte[] requestServerIdBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItem + Constants.$request_server_id, false, null);
            //为什么current_server需要设置为requestServerBytes
            //serverId==currentServerId
            if (requestServerIdBytes != null && currentServerIdBytes != null && serverId.equals(new String(currentServerIdBytes)) == true) {
                this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItem + Constants.$current_server_id, requestServerIdBytes, -1);
                this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItem + Constants.$request_server_id, null, -1);
                isModify = true;
            }
        }
        if (isModify == true) {
            this.updateTaskItemFlag(taskType);
        }
    }


    public int getTaskItemCount(String taskType) throws Exception {
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String taskItemsPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems;
        return this.getZooKeeper().getChildren(taskItemsPath, false).size();
    }

    public void clearExpireTaskTypeRunningInfo(String baseTaskType, String serverId, double expireDateInternal) throws Exception {
        for (String taskType : this.getZooKeeper().getChildren(this.baseTaskTypePath + Constants.slash + baseTaskType, false)) {
            String taskItemsPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems;
            Stat stat = this.getZooKeeper().exists(taskItemsPath, false);//下面所有的节点均有效
            if (stat == null || zooKeeperCurrentTimeMillis() - stat.getMtime() > (long) (expireDateInternal * 24 * 3600 * 1000)) {
                ZooKeepers.directoryStructureDelete(this.getZooKeeper(), this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType);
            }
        }
    }


    public int clearExpireServers(String taskType, long expireTime) throws Exception {
        int count = 0;
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String serversPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.servers;
        if (this.getZooKeeper().exists(serversPath, false) == null) {
            String taskTypePath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType;// ../
            if (this.getZooKeeper().exists(taskTypePath, false) == null) {
                this.getZooKeeper().create(taskTypePath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            }
            //
            this.getZooKeeper().create(serversPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        for (String server : this.getZooKeeper().getChildren(serversPath, false)) {
            try {
                Stat stat = this.getZooKeeper().exists(serversPath + Constants.slash + server, false);
                if (zooKeeperCurrentTimeMillis() - stat.getMtime() > expireTime) {//删除过期的server
                    ZooKeepers.directoryStructureDelete(this.getZooKeeper(), serversPath + Constants.slash + server);
                    count++;
                }
            } catch (Exception e) {

                count++;
            }
        }
        return count;
    }


    public int createTaskItemIfNotFind(String taskType,
                                       List<String> serverIdList) throws Exception//clear current_server
    {
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String taskItemsPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems;

        int count = 0;
        for (String taskItem : this.getZooKeeper().getChildren(taskItemsPath, false)) {
            byte[] currentServerIdBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItem + Constants.$current_server_id, false, null);
            if (currentServerIdBytes != null) {
                String currentServerId = new String(currentServerIdBytes);
                boolean isFind = false;
                for (String serverId : serverIdList) {
                    if (currentServerId.equals(serverId))//符合taskItem的server都清空
                    {
                        isFind = true;
                        break;
                    }
                }
                if (isFind == false)//不存在则创建
                {
                    this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItem + Constants.$current_server_id, null, -1);
                    count = count + 1;
                }
            } else {
                count = count + 1;
            }
        }
        return count;
    }

    public List<String> loadServerIdList(String taskType) throws Exception {
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String serversPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.servers;
        if (this.getZooKeeper().exists(serversPath, false) == null) {
            return new ArrayList<String>();
        }
        List<String> serverIdList = this.getZooKeeper().getChildren(serversPath, false);
        Collections.sort(serverIdList, new Comparator<String>() {
            public int compare(String leftString, String rightString) {
                return leftString.substring(leftString.lastIndexOf("$") + 1).compareTo(
                        rightString.substring(rightString.lastIndexOf("$") + 1));
            }
        });
        return serverIdList;
    }


    public List<TaskHandler> getTaskHandlerList(String taskType)
            throws Exception {
        List<TaskHandler> taskHandlerList = new ArrayList<TaskHandler>();
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String serversPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.servers;
        if (this.getZooKeeper().exists(serversPath, false) == null) {
            return taskHandlerList;
        }
        List<String> serverIdList = this.getZooKeeper().getChildren(serversPath, false);
        Collections.sort(serverIdList, new Comparator<String>() {
            public int compare(String leftString, String rightString) {
                return leftString.substring(leftString.lastIndexOf(Constants.$) + 1).compareTo(
                        rightString.substring(rightString.lastIndexOf(Constants.$) + 1));
            }
        });
        for (String serverId : serverIdList) {
            try {
                String taskHandlerBytes = new String(this.getZooKeeper().getData(serversPath + Constants.slash + serverId, false, null));
                TaskHandler taskHandler = this.gson.fromJson(taskHandlerBytes, TaskHandler.class);
                taskHandler.setCenterServerTime(new Timestamp(this.zooKeeperCurrentTimeMillis()));
                taskHandlerList.add(taskHandler);
            } catch (Exception e) {
                log.debug(e.getMessage(), e);
            }
        }
        return taskHandlerList;
    }

    /**
     * 一个服务器上面挂载很多TaskHandler
     */
    public List<TaskHandler> getTaskHandlerListByMachineId(String machineId)
            throws Exception {
        List<TaskHandler> taskHandlerList = new ArrayList<TaskHandler>();
        for (String baseTaskType : this.getZooKeeper().getChildren(this.baseTaskTypePath, false)) {
            for (String taskType : this.getZooKeeper().getChildren(this.baseTaskTypePath + Constants.slash + baseTaskType, false)) {
                String serversPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.servers;
                for (String serverId : this.getZooKeeper().getChildren(serversPath, false)) {
                    String taskHandlerString = new String(this.getZooKeeper().getData(serversPath + Constants.slash + serverId, false, null));
                    TaskHandler taskHandler = this.gson.fromJson(taskHandlerString, TaskHandler.class);
                    taskHandler.setCenterServerTime(new Timestamp(this.zooKeeperCurrentTimeMillis()));
                    if (taskHandler.getMachineId().equals(machineId))//全部遍历 找到machineId相关的task handler
                    {
                        taskHandlerList.add(taskHandler);
                    }
                }
            }
        }
        Collections.sort(taskHandlerList, new Comparator<TaskHandler>() {
            public int compare(TaskHandler leftTaskHandler, TaskHandler rightTaskHandler) {
                int result = leftTaskHandler.getTaskType().compareTo(rightTaskHandler.getTaskType());
                if (result == 0) {
                    String leftServer = leftTaskHandler.getServerId();
                    String rightServer = rightTaskHandler.getServerId();
                    result = leftServer.substring(leftServer.lastIndexOf(Constants.$) + 1).compareTo(
                            rightServer.substring(rightServer.lastIndexOf(Constants.$) + 1));
                }
                return result;
            }
        });
        return taskHandlerList;
    }

    public boolean isLeaderServerId(String serverId, List<String> serverIdList) {
        return serverId.equals(getLeaderServerId(serverIdList));
    }

    /**
     * serverIdList中最小的那个
     */
    public String getLeaderServerId(List<String> serverIdList) {
        if (serverIdList == null || serverIdList.size() == 0) {
            return null;
        }
        long num = Long.MAX_VALUE;
        long $num = -1;
        String leaderServerId = null;
        for (String serverId : serverIdList) {
            $num = Long.parseLong(serverId.substring(serverId.lastIndexOf(Constants.$) + 1));//serverName $之后为index
            if (num > $num) {//min
                num = $num;
                leaderServerId = serverId;
            }
        }
        return leaderServerId;
    }



    public void assignTaskItem(String taskType,//
                               String currentUuid, //
                               int maxTaskItemsOfOneThreadGroup,//
                               List<String> serverIdList) throws Exception {
        if (this.isLeaderServerId(currentUuid, serverIdList) == false) {
            return;
        }
        if (serverIdList.size() <= 0) {
            return;
        }
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String taskItemsPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.taskItems;
        List<String> taskItemList = this.getZooKeeper().getChildren(taskItemsPath, false);

        Collections.sort(taskItemList, new Comparator<String>() {
            public int compare(String leftString, String rightString) {
                if (StringUtils.isNumeric(leftString) && StringUtils.isNumeric(rightString)) {
                    int leftIntValue = Integer.parseInt(leftString);
                    int rightIntValue = Integer.parseInt(rightString);
                    if (leftIntValue == rightIntValue) {
                        return 0;
                    } else if (leftIntValue > rightIntValue) {
                        return 1;
                    } else {
                        return -1;
                    }
                } else {
                    return leftString.compareTo(rightString);
                }
            }
        });
        int unModifyCount = 0;
        int[] taskAllot = Schedules.allot(taskItemList.size(), serverIdList.size());//taskItemList size/serverId num
        int index = 0;
        int accumulative = 0;

        for (int i = 0; i < taskItemList.size(); i++) {

            if (index < serverIdList.size() //
                    && //
                    i >= accumulative + taskAllot[index])//index from 0,so index++
            {
                accumulative = accumulative + taskAllot[index];
                index = index + 1;
            }//递增index+1
            //index serverId
            String noServerId = "";
            String serverId = noServerId;
            if (index < serverIdList.size()) {
                serverId = serverIdList.get(index);
            }
            //
            String taskItem = taskItemList.get(i);
            byte[] currentServerBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItem + Constants.$current_server_id, false, null);
            byte[] requestServerBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItem + Constants.$request_server_id, false, null);

            if (currentServerBytes == null || new String(currentServerBytes).equals(noServerId))//""
            {
                this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItem + Constants.$current_server_id, serverId.getBytes(), -1);
                this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItem + Constants.$request_server_id, null, -1);
            } else if (new String(currentServerBytes).equals(serverId) == true && requestServerBytes == null) {
                unModifyCount = unModifyCount + 1;
            } else {// set(new String(currentServerBytes).equals(serverId) == true && requestServerBytes != null) wher(new String(currentServerBytes).equals(serverId) == true && requestServerBytes == null)
                this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItem + Constants.$request_server_id, serverId.getBytes(), -1);
            }
        }

        if (unModifyCount < taskItemList.size()) {
            log.info("updateTaskItemFlag......" + taskType + "  ,currentUuid " + currentUuid);

            this.updateTaskItemFlag(taskType);//why unModifyCount < taskItemList.size() ,need reload
        }
        if (log.isDebugEnabled()) {
            StringBuffer sb = new StringBuffer();
            for (TaskItem taskItem : this.loadTaskItemList(taskType)) {
                sb.append("\n").append(taskItem.toString());
            }
            log.debug(sb.toString());
        }
    }

    public void registerTaskHandler(@NonNull TaskHandler taskHandler) throws Exception {

        if (taskHandler.isRegister() == true) {
            throw new Exception(taskHandler.getServerId() + " ");
        }
        String taskTypePath = this.baseTaskTypePath + Constants.slash + taskHandler.getBaseTaskType() + Constants.slash + taskHandler.getTaskType();
        if (this.getZooKeeper().exists(taskTypePath, false) == null) {
            this.getZooKeeper().create(taskTypePath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        //
        String serversPath = taskTypePath + Constants.slash + this.servers;
        if (this.getZooKeeper().exists(serversPath, false) == null) {
            this.getZooKeeper().create(serversPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        //
        String serverId = Uuids.getUuid();//new uuid
        String serverPath = serversPath + Constants.slash + taskHandler.getTaskType() + Constants.$ + taskHandler.getIp() + Constants.$ + serverId + Constants.$;
        this.getZooKeeper().create(serverPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT_SEQUENTIAL);
        taskHandler.setServerId(serverId);//到底是uuid还是  ip+uuid

        Timestamp heartBeatTime = new Timestamp(this.zooKeeperCurrentTimeMillis());
        taskHandler.setHeartBeatTime(heartBeatTime);

        String value = this.gson.toJson(taskHandler);
        this.getZooKeeper().setData(serverPath, value.getBytes(), -1);//taskHandler注册在serverPath
        taskHandler.setRegister(true);
    }

    public boolean refreshHeartBeatTimeAndVersion(TaskHandler taskHandler) throws Exception {
        Timestamp heartBeatTime = new Timestamp(this.zooKeeperCurrentTimeMillis());
        String serverIdPath = this.baseTaskTypePath + Constants.slash + taskHandler.getBaseTaskType() + Constants.slash + taskHandler.getTaskType()
                + Constants.slash + this.servers + Constants.slash + taskHandler.getServerId();
        if (this.getZooKeeper().exists(serverIdPath, false) == null) {
            taskHandler.setRegister(false);
            return false;
        } else {
            Timestamp oldHeartBeatTime = taskHandler.getHeartBeatTime();
            taskHandler.setHeartBeatTime(heartBeatTime);
            taskHandler.setVersion(taskHandler.getVersion() + 1);
            String value = this.gson.toJson(taskHandler);
            try {
                this.getZooKeeper().setData(serverIdPath, value.getBytes(), -1);
                return true;
            } catch (Exception e) {
                taskHandler.setHeartBeatTime(oldHeartBeatTime);
                taskHandler.setVersion(taskHandler.getVersion() - 1);//为什么-1,这样用来恢复原来的记录
                throw e;
            }

        }
    }

    /**
     * TaskHandler就是挂在taskType 下面的servers下面
     */
    public void unRegisterTaskHandler(String taskType, String serverId) throws Exception//delete path
    {
        String baseTaskType = Schedules.getBaseTaskType(taskType);
        String serverIdPath = this.baseTaskTypePath + Constants.slash + baseTaskType + Constants.slash + taskType + Constants.slash + this.servers + Constants.slash + serverId;
        if (this.getZooKeeper().exists(serverIdPath, false) != null) {
            this.getZooKeeper().delete(serverIdPath, -1);
        }
    }

    /**
     * 相应的TaskHandler都会有同样的操作
     */

    public void pauseTaskType(String baseTaskType) throws Exception {
        TaskType taskType = this.loadBaseTaskType(baseTaskType);
        taskType.setStatus(TaskType.status_pause);
        this.updateBaseTaskType(taskType);
    }

    /**
     * 相应的TaskHandler都会有同样的操作
     */

    public void resumeTaskType(String baseTaskType) throws Exception {
        TaskType taskType = this.loadBaseTaskType(baseTaskType);
        taskType.setStatus(TaskType.status_resume);
        this.updateBaseTaskType(taskType);
    }


    public long zooKeeperCurrentTimeMillis() {
        return (this.zooKeeperBaseDateTimeLong - this.localBaseDateTimeLong) + System.currentTimeMillis();//求的是zk时间
    }

}



 